package events

import (
	"context"
	"github.com/IBM/sarama"
	"jk-time/webook/feed/domain"
	"jk-time/webook/feed/service"
	"jk-time/webook/pkg/logger"
	"jk-time/webook/pkg/saramax"
	"time"
)

type FeedEvent struct {
	Type     string
	MetaData map[string]string
}
type Consume interface {
}
type FeedEventConsumer struct {
	client sarama.Client
	l      logger.Logger
	svc    service.FeedService
}

func (FeedEvent) Topic() string {
	return "feed"
}
func NewFeedEventConsumer(client sarama.Client, l logger.Logger, svc service.FeedService) *FeedEventConsumer {
	return &FeedEventConsumer{client: client, l: l, svc: svc}
}
func (k *FeedEventConsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("feed_base", k.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{FeedEvent{}.Topic()},
			saramax.NewHandler[FeedEvent]("feed_base", k.Consume))
		if err != nil {
			k.l.Error("退出了消费循环,异常", logger.Error(err))
		}
	}()
	return err
}
func (k *FeedEventConsumer) StartBatch() error {
	return nil
}
func (k *FeedEventConsumer) Consume(msg *sarama.ConsumerMessage, t FeedEvent) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	k.l.Info("kafka:", logger.String("topic", t.Topic()),
		logger.String("value", string(msg.Value)))

	return k.svc.CreateFeedEvent(ctx, domain.FeedEvent{
		Type: t.Type,
		Ext:  t.MetaData,
	})
}
